package com.dmp.storm.topology;

import backtype.storm.Config;
import backtype.storm.LocalCluster;
import backtype.storm.StormSubmitter;
import backtype.storm.generated.AlreadyAliveException;
import backtype.storm.generated.AuthorizationException;
import backtype.storm.generated.InvalidTopologyException;
import backtype.storm.generated.StormTopology;
import backtype.storm.grouping.CustomStreamGrouping;
import backtype.storm.topology.BoltDeclarer;
import backtype.storm.topology.IRichBolt;
import backtype.storm.topology.IRichSpout;
import backtype.storm.topology.TopologyBuilder;
import backtype.storm.tuple.Fields;

import java.io.IOException;
import java.lang.reflect.Method;
import java.util.HashMap;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Properties;
import java.util.UUID;

/**
 * @author johnguo
 * @version 1.0
 * 2015.04.05
 */
public abstract class BaseTopologyCtx implements ITopologyCtx {

	public enum SS {
		LOCAL,
		CUSTOM
	}

	private TopologyBuilder builder;
	private Map<String, Object> componentCollection;

	public BaseTopologyCtx() {
		builder = new TopologyBuilder();
		componentCollection = new HashMap<String, Object>();
	}

	/**
	 * 通过set方法给对象增加属性
	 * @param methodName
	 * @param obj 添加属性值的对象
	 * @param prop 属性键值对
	 * @throws Exception
	 * @author johnguo
	 */
	protected void addProperty(String methodName, Object obj, Object prop){
		try{
			Method[] methods = obj.getClass().getMethods();
			for (Method method : methods) {
				if (methodName.equals(method.getName())) {
					Class<?>[] clazz = method.getParameterTypes();
					String type = clazz[0].getName();
					if (type.equals("java.lang.String")) {
						method.invoke(obj, (String) prop);
						return;
					}
					if (type.equals("[Ljava.lang.String;")) {
						Object props = ((String) prop).split(",");
						method.invoke(obj, props);
						return;
					}
					if (type.equals(int.class) || type.equals("java.lang.Integer")) {
						method.invoke(obj, Integer.valueOf((String) prop));
						return;
					}
					if (type.equals(byte.class) || type.equals("java.lang.Byte")) {
						method.invoke(obj, Byte.valueOf((String) prop));
						return;
					}
					if (type.equals(short.class) || type.equals("java.lang.Short")) {
						method.invoke(obj, Short.valueOf((String) prop));
						return;
					}
					if (type.equals(long.class) || type.equals("java.lang.Long")) {
						method.invoke(obj, Long.valueOf((String) prop));
						return;
					}
					if (type.equals(float.class) || type.equals("java.lang.Float")) {
						method.invoke(obj, Float.valueOf((String) prop));
						return;
					}
					if (type.equals(double.class)
							|| type.equals("java.lang.Double")) {
						method.invoke(obj, Double.valueOf((String) prop));
						return;
					}
				}
			}
		} catch (Exception e){
			System.err.println("注入参数异常");
			e.printStackTrace();
			System.exit(-1);
		}
	}

	/*
	 * 创建拓扑
	 * 
	 * @see com.asiainfo.storm.ITopologyCtx#createTopology()
	 */
	@Override
	public StormTopology createTopology() {
		StormTopology topology = builder.createTopology();
		return topology;
	}

	protected Object createObject(String classPath, Map<String, String> props) {
		Object obj = null;
		try {
			obj = Class.forName(classPath).newInstance();
			if(props != null) {
				for (Entry<String, String> entry : props.entrySet()) {
					String attrKey = entry.getKey();
					String attrValue = entry.getValue();
					this.addProperty("set" + attrKey.substring(0, 1).toUpperCase()
							+ attrKey.substring(1), obj, attrValue);
				}
			}
		} catch (ClassNotFoundException e) {
			System.err.println("无法创建类" + classPath);
			e.printStackTrace();
			System.exit(-1);
		} catch (InstantiationException e) {
			e.printStackTrace();
			System.exit(-1);
		} catch (IllegalAccessException e) {
			e.printStackTrace();
			System.exit(-1);
		}
		return obj;
	}

	protected abstract void createWorkFlow(BaseTopologyCtx.SS ss, String name);

	protected void createStreamGrouping(String name, String classPath,
										Map<String, String> props) {
		CustomStreamGrouping groupComponent = (CustomStreamGrouping) this
				.createObject(classPath, props);
		componentCollection.put(name, groupComponent);
	}

	protected void createSpout(String name, String classPath,
							   Map<String, String> props, String num) {
		IRichSpout richSpout = (IRichSpout) this.createObject(classPath, props);
		builder.setSpout(name, richSpout,
				num == null ? 1 : Integer.valueOf(num));
		componentCollection.put(name, richSpout);
	}

	protected void createBolt(String name, String classPath,
							  Map<String, String> props, String num, String grouptype,
							  String group) {
		IRichBolt richBolt = (IRichBolt) this.createObject(classPath, props);
		BoltDeclarer boltDeclarer = builder.setBolt(name, richBolt,
				num == null ? 1 : Integer.valueOf(num));
		componentCollection.put(name, richBolt);
		if ("shuffle".equals(grouptype)) {
			boltDeclarer.shuffleGrouping(group);
			return;
		}
		if ("fields".equals(grouptype)) {
			boltDeclarer.fieldsGrouping(group,new Fields("keyword"));
			return;
		}
		if ("direct".equals(grouptype)) {
			boltDeclarer.directGrouping(group);
			return;
		}
		if ("custom".equals(grouptype)) {
			String[] groups = group.split(",");
			try {
				CustomStreamGrouping customStreamGrouping = (CustomStreamGrouping) componentCollection
						.get(groups[1]);
				boltDeclarer.customGrouping(groups[0], customStreamGrouping);
			} catch (Exception e) {
				System.err.println("加载自定义分组时发生异常");
				e.printStackTrace();
				System.exit(-1);
			}
			return;
		}
	}
	@Override
	public void startLocal(String topologyName, Config conf){
		LocalCluster local = new LocalCluster();
		local.submitTopology(topologyName, conf, this.createTopology());
	}
	@Override
	public void startLocal(String topologyName){
		Config conf = new Config();
		this.startLocal(topologyName, conf);
	}
	@Override
	public void startLocal(Config conf){
		this.startLocal(UUID.randomUUID().toString(), conf);
	}
	@Override
	public void startLocal(){
		Config conf = new Config();
		this.startLocal(conf);
	}
	@Override
	public void startLocalByPropFile(String filepath) throws IOException {
		Properties prop = new Properties();
		prop.load(this.getClass().getClassLoader().getResourceAsStream(filepath));
		LocalCluster local = new LocalCluster();
		local.submitTopology(UUID.randomUUID().toString(), prop, this.createTopology());
	}
	@Override
	public void start(String topologyName, Config conf) throws AlreadyAliveException, InvalidTopologyException, AuthorizationException{
		StormSubmitter.submitTopology(topologyName, conf, this.createTopology());
	}
	@Override
	public void start(String topologyName) throws AlreadyAliveException, InvalidTopologyException, AuthorizationException{
		Config conf = new Config();
		int cnum = componentCollection.size();
		conf.setNumWorkers(cnum/10+cnum%10>5?1:0);
		conf.setNumAckers(cnum/50+cnum%50>25?1:0);
		this.start(topologyName, conf);
	}
	@Override
	public void start(Config conf) throws AlreadyAliveException, InvalidTopologyException, AuthorizationException{
		this.start(UUID.randomUUID().toString(), conf);
	}
	@Override
	public void start() throws AlreadyAliveException, InvalidTopologyException, AuthorizationException{
		this.start(UUID.randomUUID().toString());
	}
	@Override
	public void startByPropFile(String filepath) throws AlreadyAliveException, InvalidTopologyException, IOException, AuthorizationException{
		Properties prop = new Properties();
		prop.load(this.getClass().getClassLoader().getResourceAsStream(filepath));
		StormSubmitter.submitTopology(UUID.randomUUID().toString(), prop, this.createTopology());
	}
}
